package com.example.mqtt.server.netty;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.CharsetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * 对MQTT客户端发送消息后，处理的返回消息，基于MQTT协议的，需要MQTT协议的主要内容
 */
public class MqttMsgBack {

    private static final Logger log = LoggerFactory.getLogger(MqttMsgBack.class);

    /**
     * topic -> 订阅者
     */
    private static final Map<String, Set<ChannelHandlerContext>> topicSubscribers = new HashMap<>();

    /**
     * 私有构造方法
     */
    private MqttMsgBack() {
        throw new AssertionError();
    }

    /**
     * 确认连接请求
     *
     * @param channel     通道
     * @param mqttMessage 连接消息
     */
    public static void connectAck(Channel channel, MqttMessage mqttMessage) {
        // 获取连接消息
        MqttConnectMessage mqttConnectMessage = (MqttConnectMessage) mqttMessage;

        // 固定头
        MqttFixedHeader mqttFixedHeaderInfo = mqttConnectMessage.fixedHeader();
        // 可变头
        MqttConnectVariableHeader variableHeaderInfo = mqttConnectMessage.variableHeader();

        boolean sessionPresent = false;
        if (null != variableHeaderInfo) {
            sessionPresent = variableHeaderInfo.isCleanSession();
        }

        MqttConnectPayload payload = mqttConnectMessage.payload();
        log.info("连接负载信息 {}", payload);
        String userName = payload.userName();
        String password = new String(payload.passwordInBytes(), CharsetUtil.UTF_8);

        // 验证失败
        if ("test".equals(userName) && "123456".equals(password)) {
            log.info("用户认证通过");
        } else {
            log.error("用户认证失败");
            return;
        }

        // 构建返回报文, 可变报头
        MqttConnAckMessage connAck = getMqttConnAckMessage(sessionPresent, mqttFixedHeaderInfo);
        log.info("确认连接信息 {}", connAck);
        channel.writeAndFlush(connAck);
    }

    /**
     * 构建返回报文MqttConnAckMessage
     */
    private static MqttConnAckMessage getMqttConnAckMessage(boolean sessionPresent, MqttFixedHeader mqttFixedHeaderInfo) {
        MqttConnAckVariableHeader mqttConnAckVariableHeaderBack = new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_ACCEPTED, sessionPresent);
        //	构建返回报文， 固定报头
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.CONNACK, mqttFixedHeaderInfo.isDup(), MqttQoS.AT_MOST_ONCE, mqttFixedHeaderInfo.isRetain(), 0x02);
        // 构建CONNACK消息体
        return new MqttConnAckMessage(mqttFixedHeaderBack, mqttConnAckVariableHeaderBack);
    }

    /**
     * 根据qos发布确认
     *
     * @param channel     通道
     * @param mqttMessage publish消息
     */
    public static void publishAck(Channel channel, MqttMessage mqttMessage) {
        // 读取payload的数据
        MqttPublishMessage mqttPublishMessage = (MqttPublishMessage) mqttMessage;
        MqttFixedHeader mqttFixedHeaderInfo = mqttPublishMessage.fixedHeader();
        byte[] headBytes = new byte[mqttPublishMessage.payload().readableBytes()];
        mqttPublishMessage.payload().readBytes(headBytes);
        String data = new String(headBytes);
        log.info("publish数据 {}", data);

        // 分发消息给所有订阅者
        handlePublish(mqttPublishMessage);

        MqttQoS qos = mqttFixedHeaderInfo.qosLevel();
        switch (qos) {
            //	至多一次
            case AT_MOST_ONCE:
                break;
            //	至少一次
            case AT_LEAST_ONCE:
                //	构建返回报文， 可变报头
                MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack = MqttMessageIdVariableHeader.from(mqttPublishMessage.variableHeader().packetId());
                //	构建返回报文， 固定报头
                MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBACK, mqttFixedHeaderInfo.isDup(), MqttQoS.AT_MOST_ONCE, mqttFixedHeaderInfo.isRetain(), 0x02);
                //	构建PUBACK消息体
                MqttPubAckMessage pubAck = new MqttPubAckMessage(mqttFixedHeaderBack, mqttMessageIdVariableHeaderBack);
                log.info("QoS1下发确认信息 {}", pubAck);
                channel.writeAndFlush(pubAck);
                break;
            //	刚好一次
            case EXACTLY_ONCE:
                //	构建返回报文， 固定报头
                MqttFixedHeader mqttFixedHeaderBack2 = new MqttFixedHeader(MqttMessageType.PUBREC, false, MqttQoS.AT_LEAST_ONCE, false, 0x02);
                //	构建返回报文， 可变报头
                MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack2 = MqttMessageIdVariableHeader.from(mqttPublishMessage.variableHeader().packetId());
                MqttMessage mqttMessageBack = new MqttMessage(mqttFixedHeaderBack2, mqttMessageIdVariableHeaderBack2);
                log.info("QoS2下发PUBREC信息 {}", mqttMessageBack);
                channel.writeAndFlush(mqttMessageBack);
                break;
            default:
                break;
        }
    }

    /**
     * 将消息分发给所有订阅该主题的客户端
     *
     * @param msg publish消息
     */
    private static void handlePublish(MqttPublishMessage msg) {
        String topic = msg.variableHeader().topicName();

        // 将消息分发到所有订阅者
        Set<ChannelHandlerContext> subscribers = topicSubscribers.get(topic);
        if (subscribers != null) {
            for (ChannelHandlerContext subscriber : subscribers) {
                subscriber.writeAndFlush(msg.retain()); // 发送消息给每个订阅者
            }
        }
    }

    /**
     * 发布完成qos2
     *
     * @param channel     通道
     * @param mqttMessage QoS2客户端->服务端PUBREL第三个报文：
     */
    public static void pubComp(Channel channel, MqttMessage mqttMessage) {
        MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage.variableHeader();
        //	构建返回报文， 固定报头
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBCOMP, false, MqttQoS.AT_MOST_ONCE, false, 0x02);
        //	构建返回报文， 可变报头
        MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
        MqttMessage mqttMessageBack = new MqttMessage(mqttFixedHeaderBack, mqttMessageIdVariableHeaderBack);
        log.info("QoS2下发PUBCOMP信息 {}", mqttMessageBack);
        channel.writeAndFlush(mqttMessageBack);
    }

    /**
     * 订阅确认
     *
     * @param channelHandlerContext channelHandler上下文
     * @param channel 通道
     * @param mqttMessage 订阅消息
     */
    public static void subscribeAck(ChannelHandlerContext channelHandlerContext, Channel channel, MqttMessage mqttMessage) {
        // 获取订阅消息
        MqttSubscribeMessage mqttSubscribeMessage = (MqttSubscribeMessage) mqttMessage;
        MqttMessageIdVariableHeader messageIdVariableHeader = mqttSubscribeMessage.variableHeader();

        List<String> topics = mqttSubscribeMessage.payload().topicSubscriptions().stream().map(MqttTopicSubscription::topicName).collect(Collectors.toList());
        log.info("Topic列表{}", topics);
        List<Integer> grantedQoSLevels = new ArrayList<>(topics.size());
        for (int i = 0; i < topics.size(); i++) {
            // 保存每个topic的订阅者
            String topic = mqttSubscribeMessage.payload().topicSubscriptions().get(i).topicName();
            topicSubscribers
                    .computeIfAbsent(topic, k -> ConcurrentHashMap.newKeySet())
                    .add(channelHandlerContext);
            grantedQoSLevels.add(mqttSubscribeMessage.payload().topicSubscriptions().get(i).qualityOfService().value());
        }

        //	构建返回报文， 可变报头
        MqttMessageIdVariableHeader variableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
        //	构建返回报文	有效负载
        MqttSubAckPayload payloadBack = new MqttSubAckPayload(grantedQoSLevels);
        //	构建返回报文	固定报头
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE, false, 2 + topics.size());
        //	构建返回报文	订阅确认
        MqttSubAckMessage subAck = new MqttSubAckMessage(mqttFixedHeaderBack, variableHeaderBack, payloadBack);
        log.info("订阅连接信息 {}", subAck);
        channel.writeAndFlush(subAck);
    }

    /**
     * 取消订阅确认
     *
     * @param channelHandlerContext channelHandler上下文
     * @param channel     通道
     * @param mqttMessage 取消订阅消息
     */
    public static void unsubscribeAck(ChannelHandlerContext channelHandlerContext, Channel channel, MqttMessage mqttMessage) {
        // 获取取消订阅消息
        MqttUnsubscribeMessage mqttUnsubscribeMessage = (MqttUnsubscribeMessage) mqttMessage;
        MqttMessageIdVariableHeader messageIdVariableHeader = mqttUnsubscribeMessage.variableHeader();

        List<String> topics = mqttUnsubscribeMessage.payload().topics();
        log.info("Topic列表{}", topics);
        for (String topic : topics) {
            topicSubscribers
                    .computeIfAbsent(topic, k -> ConcurrentHashMap.newKeySet())
                    .stream().filter(ctx -> channelHandlerContext != ctx);
        }

        // 构建返回报文 可变报头
        MqttMessageIdVariableHeader variableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
        // 构建返回报文 固定报头
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, MqttQoS.AT_MOST_ONCE, false, 2);

        // 构建返回报文 取消订阅确认
        MqttUnsubAckMessage unSubAck = new MqttUnsubAckMessage(mqttFixedHeaderBack, variableHeaderBack);
        log.info("取消订阅确认{}", unSubAck);
        channel.writeAndFlush(unSubAck);
    }

    /**
     * 心跳响应
     *
     * @param channel 通道
     */
    public static void pingResp(Channel channel) {
        //	心跳响应报文	11010000 00000000  固定报文
        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PINGRESP, false, MqttQoS.AT_MOST_ONCE, false, 0);
        MqttMessage mqttMessageBack = new MqttMessage(fixedHeader);
        log.info("心跳响应 {}", mqttMessageBack);
        channel.writeAndFlush(mqttMessageBack);
    }
}
